package com.gitee.fsmxm.thinking.in.concurrent.basic;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 目前操作：使用线程并行优化串行操作
 * <p>获取未对账订单
 * <p>获取派送订单
 * <p>执行对账操作
 * <p>差异写入差异库
 */
public class _09_CyclicBarrier_OrdersSample {

    public static class Order {
        int orderId;

        public Order(int id) {
            orderId = id;
        }
    }

    public static class OrderService {
        // 获取未对账订单
        public List<Order> getPOrders() {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Arrays.asList(new Order(1), new Order(2));
        }

        // 获取派送订单
        public List<Order> getDOrders() {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Arrays.asList(new Order(1), new Order(2));
        }

        // 执行对账操作
        public Object check(List<Order> pos, List<Order> dos) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return new Object();
        }

        // 差异写入差异库
        public void save(Object diff) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static final OrderService service = new OrderService();

    // 获取未对账订单和获取派送订单并行，执行对账操作时再并行获取下一次的订单，并使用线程池和CyclicBarrier
    private static void testInThreadPool() throws InterruptedException {
        final ExecutorService executor = Executors.newFixedThreadPool(2);
        final ExecutorService executorForBarrier = Executors.newFixedThreadPool(1);
        AtomicReference<List<Order>> pOrders = new AtomicReference<>();
        AtomicReference<List<Order>> dOrders = new AtomicReference<>();
        AtomicLong t1 = new AtomicLong(System.currentTimeMillis());
        CyclicBarrier barrier = new CyclicBarrier(2, () -> {
            executorForBarrier.execute(() -> {
                Object diff = service.check(pOrders.get(), dOrders.get());
                service.save(diff);
                System.out.println("并行操作，耗时： " + (System.currentTimeMillis() - t1.get()) + "毫秒");
                t1.set(System.currentTimeMillis());
            });
        });

        executor.execute(() -> {
            while (true) {
                pOrders.set(service.getPOrders());
                try {
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        });
        executor.execute(() -> {
            while (true) {
                dOrders.set(service.getDOrders());
                try {
                    barrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public static void main(String[] args) throws InterruptedException {
        testInThreadPool();
    }

}
